package com.changhong.disruptorsource.quickstart.multi;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.concurrent.Executors;

public class MainNew {

    public static void main(String[] args) throws InterruptedException {
        long beginTime = System.currentTimeMillis();
        int count = 10000000;
        //1 创建RingBuffer
        RingBuffer<Order> ringBuffer =
                RingBuffer.create(ProducerType.SINGLE,
                        () -> new Order(),
                            1024,
                        new YieldingWaitStrategy());

        //2 通过ringBuffer 创建一个屏障
        SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();

        //3 创建多个消费者数组:
        Consumer[] consumers = new Consumer[1000];
        for (int i = 0; i < consumers.length; i++) {
            consumers[i] = new Consumer("C" + i);
        }

        //4 构建多消费者工作池
        WorkerPool<Order> workerPool = new WorkerPool<Order>(
                ringBuffer,
                sequenceBarrier,
                new EventExceptionHandler(),
                consumers);

        //5 设置多个消费者的sequence序号 用于单独统计消费进度, 并且设置到ringbuffer中
        ringBuffer.addGatingSequences(workerPool.getWorkerSequences());

        //6 启动workerPool
        workerPool
                .start(Executors.newFixedThreadPool(1000));

        final Producer producer = new Producer(ringBuffer);
        for (int j = 0; j < count; j++) {
            producer.sendData(String.valueOf(j));
        }
        while (true) {
            try {
                Thread.sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            int size = Consumer.count.intValue();
            if (size == count) break;
        }
        long endTime = System.currentTimeMillis();
        System.out.println(endTime - beginTime);
    }

    static class EventExceptionHandler implements ExceptionHandler<Order> {
        public void handleEventException(Throwable ex, long sequence, Order event) {
        }

        public void handleOnStartException(Throwable ex) {
        }

        public void handleOnShutdownException(Throwable ex) {
        }
    }


}
